package com.zxl.study.listener;

import com.zxl.study.entities.CacheInvalidations;
import com.zxl.study.entities.MetaDatabase;
import com.zxl.study.entities.MetaTable;
import com.zxl.study.mapper.CacheInvalidationsMapper;
import com.zxl.study.mapper.MetaDatabaseMapper;
import com.zxl.study.mapper.MetaTableMapper;
import com.zxl.study.service.MetaDatabaseService;
import com.zxl.study.config.DatabaseConfig;
import lombok.extern.slf4j.Slf4j;
import org.postgresql.PGConnection;
import org.postgresql.PGNotification;
import org.springframework.stereotype.Service;
import tk.mybatis.mapper.entity.Example;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.List;
import java.util.Properties;

@Slf4j
@Service
public class PostgresListener {

    @Resource
    private DatabaseConfig databaseConfig;
    @Resource
    private MetaTableMapper metaTableMapper;
    @Resource
    private MetaDatabaseMapper metaDatabaseMapper;
    @Resource
    private CacheInvalidationsMapper cacheInvalidationsMapper;
    @Resource
    private MetaDatabaseService metaDatabaseService;

    /**
     * 监听数据变更通知
     *
     */
    public void listener() {
        try {
            Properties props = new Properties();
            props.setProperty("user", databaseConfig.getUsername());
            props.setProperty("password", databaseConfig.getPassword());

            Connection conn = DriverManager.getConnection(databaseConfig.getUrl(), props);
            Statement stmt = conn.createStatement();
            log.info("开始监听数据变化！");
            // 监听 'data_changes' 频道
            stmt.execute("LISTEN data_changes");
            do {
                // 检查并处理未处理的数据
                processPendingNotifications();
                // 等待通知
                PGConnection acorn = conn.unwrap(PGConnection.class);
                PGNotification[] notifications = acorn.getNotifications();

                if (notifications != null && notifications.length != 0) {
                    log.info("检测到数据变更，开始处理缓存数据");
                    for (PGNotification notification : notifications) {
                        // 处理通知
                        String[] payload = notification.getParameter().split(",");
                        String tableName = payload[0];
                        String operationType = payload[1];
                        int recordId = Integer.parseInt(payload[2]);
                        CacheInvalidations cacheInvalidations = new CacheInvalidations().setTableName(tableName)
                                .setRecordId(recordId).setOperationType(operationType)
                                .setIsProcess(Boolean.FALSE).setFailCount(0);
                        cacheInvalidationsMapper.insertSelective(cacheInvalidations);
                        processPGNotification(cacheInvalidations);
                    }
                }
                // 休眠一段时间后继续检查
                Thread.sleep(1000);
            } while (true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 检查并处理未处理的通知
     *
     */
    private void processPendingNotifications() {
        Example example = new Example(CacheInvalidations.class);
        Example.Criteria criteria = example.createCriteria();
        criteria.andEqualTo("isProcess", Boolean.FALSE);
        List<CacheInvalidations> pendingNotifications = cacheInvalidationsMapper.selectByExample(example);
        for (CacheInvalidations cacheInvalidations : pendingNotifications) {
            if (cacheInvalidations.getFailCount() < 4) {
                try {
                    processPGNotification(cacheInvalidations);
                } catch (Exception e) {
                    cacheInvalidations.setFailCount(cacheInvalidations.getFailCount() + 1);
                    cacheInvalidationsMapper.updateByPrimaryKeySelective(cacheInvalidations);
                }
            } else {
                log.error("Notification {} failed to process more than 4 times", cacheInvalidations.getId());
            }
        }
    }

    /**
     * 处理数据变更通知消息
     *
     * @param cacheInvalidations 待处理的数据变更通知消息
     */
    private void processPGNotification(CacheInvalidations cacheInvalidations) {
        MetaTable metaTable = metaTableMapper.selectByPrimaryKey(cacheInvalidations.getRecordId());
        MetaDatabase metaDatabase = metaDatabaseMapper.selectByPrimaryKey(metaTable.getDatabaseId());
        metaDatabaseService.scanRedisData(metaDatabase);
        cacheInvalidations.setIsProcess(Boolean.TRUE);
        cacheInvalidationsMapper.updateByPrimaryKeySelective(cacheInvalidations);
    }
}
